package com.test.sync.queue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by Administrator on 2017/4/5.
 * <p>
 * 模拟 Q
 * 场景：一个流程会产生消息，但量不会很大，占用内存不会很多，将消息放在一个内存队列中，然后由消费者直接消费
 */
public class QueueDemo {

    private static Logger logger = LoggerFactory.getLogger(QueueDemo.class);

    private static LinkedBlockingDeque<Integer> queue = new LinkedBlockingDeque<>();

    private static AtomicBoolean overFlag = new AtomicBoolean(false);

    private static Random random = new Random(1000);

    private static int bound = 50;


    /**
     * 生产消息
     */
    public static void producer() {
        ExecutorService service = Executors.newFixedThreadPool(1);
        Runnable work = () -> {
            for (int i = 1; i <= 1000; i++) {
                try {
                    Thread.sleep(random.nextInt(bound));
                } catch (InterruptedException e) {
                }
                logger.info("produce : " + i);
                queue.add(i);
                if (i >= 1000) {
                    overFlag.set(true);
                    break;
                }
            }
        };
        service.execute(work);
        service.shutdown();
    }

    public static void consumer() {
        ExecutorService service = Executors.newFixedThreadPool(1);
        List<Integer> result = new ArrayList<>();
        Runnable work = () -> {
            while (!overFlag.get() || queue.size() > 0) {
                // if (queue.size() <= 0) continue;
                try {
                    Thread.sleep(random.nextInt(bound));
                } catch (InterruptedException e) {
                }
                int i;
                try {
                    i = queue.pollLast();
                } catch (Exception e) {
                    logger.error(String.format("flag is %s, queue size is %s, error msg is %s", overFlag.get(), queue.size(), e.getMessage()));
                    continue;
                }
                logger.info("consumer : " + i);
                result.add(i);
            }
            result.sort((s, d) -> (Integer.parseInt(s.toString()) - Integer.parseInt(d.toString())));
            logger.info(result.toString());
        };
        service.execute(work);
        service.shutdown();
    }


    public static void main(String[] args) {
        producer();

        consumer();
    }

}
